#利用sql进行数据分析 提交到pyspark集群中
# coding:utf8
#@Author：LU80
from pyhive import hive
import pandas as pd
from sqlalchemy import create_engine
#虚拟机MySQL链接
engine = create_engine('mysql+pymysql://root:123456@192.168.88.161:3306/book_movies_olpk')

if __name__ == '__main__':
    # 获取到Hive(Spark ThriftServer的链接)
    conn = hive.Connection(host="node1", port=10000, username="root", database='book_movies_olpk')
    cursor = conn.cursor()

    #Task-1 电影播放数据分析
    cursor.execute("select name,amount from moviedata order by amount desc limit 20;")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    #list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns=['name', 'amount']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('datadb_test1', engine, index=False)
    print("Task—1 finished")

    # Task-2 首映数据分析
    cursor.execute("select premiere,count(1) as new_data from moviedata group by premiere order by new_data desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['premiere', 'new_data']
    print(df)
    df.to_sql('datadb_test2', engine, index=False)
    print("Task—2 finished")

    #Task-3  电影最高评分变化数据分析
    cursor.execute("select name,evaluate_new from moviedata where evaluate_new <10 and length(name) != 0 order by evaluate_new desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['name', 'evaluate_new']
    print(df)
    df.to_sql('datadb_test3', engine, index=False)
    print("Task—3 finished")

    #Task-4  电影最低评分变化数据分析
    cursor.execute("select name,evaluate_new from moviedata where evaluate_new <10 and length(name) != 0 order by evaluate_new limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['name', 'evaluate_new']
    print(df)
    df.to_sql('datadb_test4', engine, index=False)
    print("Task—4 finished")

    #Task-5  每年电影上映数据分析
    cursor.execute("select date_year,count(1) as count_year from moviedata group by date_year order by count_year desc limit 20; ")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['date_year', 'count_year']
    print(df)
    df.to_sql('datadb_test5', engine, index=False)
    print("Task—5 finished")

    #Task-6  制作电影最多的数据分析
    cursor.execute("select nation,count(1) as count_nat from moviedata group by nation order by count_nat desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['nation', 'count_nat']
    print(df)
    df.to_sql('datadb_test6', engine, index=False)
    print("Task—5 finished")

    #Task-7  制作电影最少的数据分析
    cursor.execute("select nation,count(1) as count_nat from moviedata group by nation having count_nat >= 5 order by count_nat limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['nation', 'count_nat']
    print(df)
    df.to_sql('datadb_test7', engine, index=False)
    print("Task—5 finished")

    # Task-8  电影类型数据分析
    cursor.execute("select tag,count(1) as count_state from moviedata group by tag order by count_state desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['tag', 'connt_siate']
    print(df)
    df.to_sql('datadb_test8', engine, index=False)
    print("Task—5 finished")

    # Task-9  电影最少类型分析
    cursor.execute("select tag,count(1) as count_state from moviedata group by tag having count_state >= 5 order by count_state  limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['tag', 'cout_tag']
    print(df)
    df.to_sql('datadb_test9', engine, index=False)
    print("Task—5 finished")

    # Task-10  电影差评数据分析
    cursor.execute("select name,evaluate from moviedata order by evaluate desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['name', 'evaluate']
    print(df)
    df.to_sql('datadb_test10', engine, index=False)
    print("Task—5 finished")

    # Task-11  求被评分次数最多的10部电影，并给出评分次数（电影名，评分次数）
    cursor.execute("select a.moviename as moviename,count(a.moviename) as total "
                   "from t_movie a join t_rating b "
                   "on a.movieid=b.movieid "
                   "group by a.moviename "
                   "order by total desc "
                   "limit 10;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['moviename', 'total']
    print(df)
    df.to_sql('datadb_test11', engine, index=False)

    # Task-12  分别求男性，女性当中评分最高的20部电影（性别，电影名，影评分）三表联合查询，按照性别过滤条件，电影名作为分组条件，影评分作为排序条件进行查询
    cursor.execute("select 'F' as sex, c.moviename as name, avg(a.rate) as avgrate, count(c.moviename) as total "
                   "from t_rating a join t_user b on a.userid=b.userid "
                   "join t_movie c on a.movieid=c.movieid "
                   "where b.sex='F' "
                   "group by c.moviename "
                   "having total >= 50 "
                   "order by avgrate desc "
                   "limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['sex', 'name','avgrate','total']
    print(df)
    df.to_sql('datadb_test12', engine, index=False)

# Task-13  求movieid = 2116这部电影各年龄段的平均影评（年龄段，影评分）
    cursor.execute("select a.age as age, avg(b.rate) as avgrate "
                   "from t_user a join t_rating b on a.userid=b.userid "
                   "where b.movieid=2116 "
                   "group by a.age;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['age', 'avgrate']
    print(df)
    df.to_sql('datadb_test13', engine, index=False)

    # Task-14   求最喜欢看电影（影评次数最多）的那位女性评最高分的20部电影（观影者，电影名，影评分）
    cursor.execute("select a.movieid as movieid, b.moviename as moviename, a.rate as rate "
                   "from t_rating a join t_movie b on a.movieid=b.movieid "
                   "where a.userid=(select a.userid "
                        "from t_rating a join t_user b on a.userid = b.userid "
                        "where b.sex='F' "
                        "group by a.userid order by count(a.userid) "
                        "desc limit 1) "
                   "order by rate "
                   "desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['movieid', 'moviename','rate']
    print(df)
    df.to_sql('datadb_test14', engine, index=False)

# Task-15  按照years=1998作为where过滤条件，按照评分作为排序条件进行查询
    cursor.execute("select a.moviename as name, a.avgrate as rate "
                   "from temp_movie_2 a "
                   "where a.years=1998 "
                   "order by rate "
                   "desc limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['name', 'rate']
    print(df)
    df.to_sql('datadb_test15', engine, index=False)